package com.telra.app.server2.service;

import com.espertech.esper.client.*;
import com.telra.app.server.data.OrderEvent;
import com.telra.app.server2.listener.AggregateOrderListener;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import java.text.DecimalFormat;

/**
 * Basic Esper Service
 */
public class EsperService {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(EsperService.class);
    private final EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
    private final static DecimalFormat fmt = new DecimalFormat("#,##0.00");

    private AggregateOrderListener listener;

    public EsperService() {
        init();
    }

    /**
     * Listen for aggregate updates from the Esper engine.  Whenever an update
     * is received by Esper an aggregate update is published.  This basically a real-time
     * group by without any timer.
     * @param listener
     */
    public void addListener(AggregateOrderListener listener) {
        this.listener = listener;
    }

    /**
     * Write an individual order to the Esper engine
     * @param order
     */
    public void writeOrder(OrderEvent order) {
        epService.getEPRuntime().sendEvent(order);
    }

    private void init()  {
        String expression = "select count(*) as count, avg(price), sum(price), max(price), category " +
                            "from com.telra.app.server.data.OrderEvent " +
                            "group by category";
        //String expression = "select count(*) as count, avg(price), sum(price), max(price), category " +
        //                    "from com.telra.app.server.data.OrderEvent.win:time(30 sec) " +
        //                    "group by category";

        EPStatement statement = epService.getEPAdministrator().createEPL(expression);
        statement.addListener(new OrderUpdateListener());
    }

    private class OrderUpdateListener implements UpdateListener {

        public void update(EventBean[] newEvents, EventBean[] oldEvents) {

            for (EventBean event : newEvents) {
                long count =  (Long)event.get("count");

                if (count > 0 && listener != null) {
                    if (log.isDebugEnabled()) {
                        log.debug(new StringBuilder().append("count=").append(event.get("count")).
                                append(", category=").append(event.get("category")).append(", sum=").
                                append(fmt.format(event.get("sum(price)"))).append(", avg=").
                                append(fmt.format(event.get("avg(price)"))).append(", max=").
                                append(fmt.format(event.get("max(price)"))).toString());
                    }
                    listener.update(
                            (OrderEvent.Category) event.get("category"),
                            ((Long) event.get("count")).intValue(),
                            (Double) event.get("avg(price)"),
                            (Double) event.get("sum(price)"),
                            (Double) event.get("max(price)")
                    );
                }
            }

        }

    }


}
